package com.tech.process

import com.tech.InitializeGroupActivity
import com.tech.common.{KafkaUtil, KuduUtil}
import com.tech.config.ApplicationConfig
import com.tech.process.nodeProcess.{BehaviorNode, NextNode, TestNode, WaitNode}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object DelayProcess {

  def delayProcess(value: String,
                   activities: ArrayBuffer[String],
                   nextNodeMap: mutable.HashMap[String, NextNode],
                   behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                   testNodeMap: mutable.HashMap[String, TestNode],
                   waitNodeMap: mutable.HashMap[String, WaitNode],
                   groupMap: mutable.HashMap[String, mutable.Set[String]]): Unit = {

    implicit val f: DefaultFormats.type = org.json4s.DefaultFormats

    val delayMap = JsonMethods.parse(value).extract[Map[String, Any]]

    val activity_id = delayMap("activity_id").toString

    //判断节点结束时是否处于活动时间内
    if (activities.contains(activity_id)) {
      //节点类型：wait behavior
      val nodeType = delayMap("node_type").toString

      nodeType match {
        case "wait" => waitProcess(delayMap, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, groupMap)
        case "behavior" => behaviorProcess(delayMap, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
      }
    }


  }

  /**
   * 等待节点时间结束，当前用户直接通过节点
   *
   * @param waitMap map
   */
  def waitProcess(waitMap: Map[String, Any],
                  nextNodeMap: mutable.HashMap[String, NextNode],
                  behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                  testNodeMap: mutable.HashMap[String, TestNode],
                  waitNodeMap: mutable.HashMap[String, WaitNode],
                  groupMap: mutable.HashMap[String, mutable.Set[String]]): Unit = {

    val project = waitMap("project").toString
    val activity_id = waitMap("activity_id").toString
    val process_id = waitMap("process_id").toString

    //判断节点
    if (waitMap.contains("customer_id") && waitMap("customer_id").toString != "") {

      val customer_id = waitMap("customer_id").toString
      val member_id = waitMap("member_id").toString

      //      KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, process_id, customer_id, member_id)
      KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)
      EventProcess.passNode(waitMap("customer_id").toString, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)

    } else if (waitMap("is_group").toString == "1") {

      val group_id = waitMap("group_id").toString

      KafkaUtil.send2Kafka4Group(ApplicationConfig.CUSTOMER_NODE_TOPIC, ApplicationConfig.PROJECT, activity_id, process_id, group_id)
      InitializeGroupActivity.passNode(project, waitMap("group_id").toString, activity_id, process_id, behaviorNodeMap, nextNodeMap, testNodeMap, waitNodeMap, groupMap)

    }

  }

  /**
   * 行为节点时间结束，当前用户经过判断后决定是否通过节点
   *
   * @param behaviorMap map
   */
  def behaviorProcess(behaviorMap: Map[String, Any],
                      nextNodeMap: mutable.HashMap[String, NextNode],
                      behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                      testNodeMap: mutable.HashMap[String, TestNode],
                      waitNodeMap: mutable.HashMap[String, WaitNode]): Unit = {

    //由于每次行为节点延时数据到达时都需要检查其是否提前通过，因此不再依赖延时节点表中发送到kafka的数据，直接从用户节点表获取

    val customer_id = behaviorMap("customer_id").toString
    val event_key = behaviorMap("event_key").toString
    val activity_id = behaviorMap("activity_id").toString
    val process_id = behaviorMap("process_id").toString
    val member_id = behaviorMap("member_id").toString

    val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

    println("rowResult != null:", rowResult != null)

    if (rowResult != null) {

      if (behaviorNodeMap(process_id).customer_behavior == 100000001) {
        //        KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, process_id, customer_id, member_id)
        KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)
        EventProcess.passNode(customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
      } else {

        val isPass = if (rowResult.isNull("limit_symbol")) {
          false
        } else {
          val limit_symbol = rowResult.getString("limit_symbol")
          limit_symbol match {
            case "between" => rowResult.getDouble("event_value") >= rowResult.getDouble("limit_value_min") && rowResult.getDouble("event_value") <= rowResult.getDouble("limit_value_max")
            case "eq" => rowResult.getDouble("event_value") == rowResult.getDouble("limit_value")
            case "neq" => rowResult.getDouble("event_value") != rowResult.getDouble("limit_value")
            case "gt" | "gte" => false
            case "lt" | "lte" => true
          }
        }

        println("isPass:", isPass)

        if (isPass) {
          //          KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, process_id, customer_id, member_id)
          KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)
          EventProcess.passNode(customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)

        }
      }

      // 将用户节点表的数据软删除
      EventProcess.deleteNode(customer_id, activity_id, process_id, event_key, System.currentTimeMillis())
    }


  }


}
